package im.composer.media.audio.stream;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueInputStream extends InputStream {
	private final BlockingQueue<byte[]> queue = new LinkedBlockingQueue<byte[]>();
	private ByteArrayInputStream current = new ByteArrayInputStream(new byte[] {});
	private boolean closed = false;

	@Override
	public int available() throws IOException {
		if (current.available() < 1) {
			if (!closed) {
				try {
					current = new ByteArrayInputStream(queue.take());
				} catch (InterruptedException e) {
					throw new IOException(e);
				}
			} else {
				byte[] b = queue.poll();
				if (b != null) {
					current = new ByteArrayInputStream(b);
				} else {
					return -1;
				}
			}
		}
		return current.available();
	}

	@Override
	public void close() throws IOException {
		closed = true;
	}

	@Override
	public int read() throws IOException {
		if (available() < 0) {
			return -1;
		}
		return current.read();
	}

	public boolean offer(byte[] e) {
		if(e==null){
			throw new NullPointerException();
		}
		if(e.length<1){
			return true;
		}
		if (closed) {
			return false;
		}
		return queue.offer(e);
	}

}
